package com.ds.lens.flink.monitor.metricsjob;

import com.ds.lens.flink.monitor.common.JacksonConfiguration;
import com.ds.lens.flink.monitor.MonitorJobContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;
@Slf4j
public class MetricsSinkFunction implements ElasticsearchSinkFunction<MetricsResult> {

    public IndexRequest createIndexRequest(MetricsResult element) {
        try{
            String indexName = MonitorJobContext.getInstance().getEsMetricdataIndex();
            String typeName = MonitorJobContext.getInstance().getEsMetricdataType();
            return Requests.indexRequest()
                    .index(indexName)
                    .type(typeName)
                    .source(JacksonConfiguration.newJsonObjectMapper().writeValueAsBytes(element), XContentType.JSON);
        }catch (JsonProcessingException e){
            log.error("MetricsSinkFunction sink failed.", e);
        }
        return null;
    }

    @Override
    public void process(MetricsResult element, RuntimeContext ctx, RequestIndexer indexer) {
        indexer.add(createIndexRequest(element));
    }


}
